use bytes::{Buf, BufMut, Bytes, BytesMut};
use core::ptr as core_ptr;

/// This is a fixed-size BytesMut Buffer.
/// |--- -|------------------|-----------------------|
/// 0----start--valid_data---end----write_space----len
pub struct FixedBuffer {
    data: BytesMut,
    // Starting position of valid data.
    start: usize,
    // Ending position of valid data.
    end: usize,
}

#[allow(dead_code)]
impl FixedBuffer {
    #[inline]
    pub fn new(capacity: usize) -> Self {
        let mut data = BytesMut::with_capacity(capacity);
        unsafe {
            data.set_len(capacity);
        }

        FixedBuffer {
            data,
            start: 0,
            end: 0,
        }
    }

    #[inline]
    pub fn write_slice(&mut self) -> &mut [u8] {
        &mut self.data[self.end..]
    }

    #[inline]
    pub fn write_slice_len(&mut self) -> usize {
        self.data.len() - self.end
    }

    #[inline]
    pub fn fill(&mut self, len: usize) {
        self.end += len;
    }

    #[inline]
    pub fn put_slice(&mut self, src: &[u8]) {
        let mut write_slice = self.write_slice();
        write_slice.put_slice(src);
        self.fill(src.len());
    }

    #[inline]
    pub fn read_slice(&self) -> &[u8] {
        &self.data[self.start..self.end]
    }

    #[inline]
    pub fn read_slice_len(&self) -> usize {
        self.end - self.start
    }

    #[inline]
    pub fn consume(&mut self, len: usize) {
        self.start += len;
        assert!(self.start <= self.end);
    }

    // 将start-end之间的数据挪移到data起始位置.
    // 比如: [used data][valid data][space]
    // 转换: [valid data][......space.....]
    #[inline]
    pub fn compaction(&mut self) {
        // 没有已消费的数据区, 直接返回即可.
        if self.start == 0 {
            return;
        }

        unsafe {
            let start = self.start;
            let end = self.end;
            let valid_size = end - start;
            let ptr = self.data.as_mut_ptr();
            core_ptr::copy(ptr.add(start), ptr, valid_size);
            self.start = 0;
            self.end = valid_size;
        }
    }

    // 注意: split_to针对的是有效数据段的偏移量, 而非data空间的偏移量.
    // 通常compaction和split_to不会混合使用.
    // 在具体的split_to之前, 我们会先advance data, 跳过已经被消费的数据.
    pub fn split_to(&mut self, at: usize) -> Bytes {
        let start = self.start;
        // split_to是对有效数据空间的拆分.
        assert!(at <= self.end - start);

        // skip alread used data.
        self.start = 0;
        self.end -= start;
        self.data.advance(start);

        self.end -= at;
        self.data.split_to(at).freeze()
    }

    // FIXME: 如果没有数据, 就不需要转成Bytes了, 这样能够提高性能.
    pub fn freeze(mut self) -> Bytes {
        // skip alread used data.
        self.data.advance(self.start);

        // set data length.
        unsafe {
            self.data.set_len(self.end - self.start);
        }

        self.data.freeze()
    }
}

#[cfg(test)]
mod tests {
    use bytes::BufMut;

    use super::FixedBuffer;

    #[test]
    pub fn test_buffer() {
        let mut buffer = FixedBuffer::new(1024);

        let mut write_slice = buffer.write_slice();
        assert!(write_slice.len() == 1024);
        write_slice.put_bytes(b'a', 128);
        buffer.fill(128);

        let read_slice = buffer.read_slice();
        assert!(read_slice.len() == 128);
        for v in read_slice {
            assert!(*v == b'a');
        }

        buffer.consume(128);
        let read_slice = buffer.read_slice();
        assert!(read_slice.len() == 0);
        let write_slice = buffer.write_slice();
        assert!(write_slice.len() == 896);

        // test split
        let mut write_slice = buffer.write_slice();
        write_slice.put_bytes(b'b', 128);
        write_slice.put_bytes(b'c', 128);
        buffer.fill(128);
        buffer.fill(128);

        let bytes = buffer.split_to(128);
        assert!(bytes.len() == 128);
        for v in bytes {
            assert!(v == b'b');
        }

        let write_slice = buffer.write_slice();
        assert!(write_slice.len() == 640);
        let read_slice = buffer.read_slice();
        assert!(read_slice.len() == 128);
        for v in read_slice {
            assert!(*v == b'c');
        }
        assert!(buffer.write_slice_len() == 640);

        let bytes = buffer.split_to(128);
        assert!(bytes.len() == 128);
        assert!(bytes.len() == 128);
        for v in bytes {
            assert!(v == b'c');
        }

        assert!(buffer.read_slice_len() == 0);
        assert!(buffer.write_slice_len() == 640);

        buffer.put_slice(b"12345678901234567890");
        buffer.consume(10);
        assert!(buffer.write_slice_len() == 620);
        buffer.compaction();
        assert!(buffer.write_slice_len() == 630);
    }
}
